package com.feidee.fd.sml.algorithm.component.validation.action

import com.feidee.fd.sml.algorithm.component.validation.ValidationParam
import com.feidee.fd.sml.algorithm.forecast.{Forecast, SparkForecast}
import com.feidee.fd.sml.algorithm.util.HttpUtil
import com.feidee.fdcommon.constant.CommonConstant
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.json4s.DefaultFormats
import org.json4s.jackson.Serialization.{read, write}

/**
  * @Author songhaicheng
  * @Date 2019/5/14 16:05
  * @Description
  * @Reviewer
  */
case class ConvertModelTypeAction(
                                   convertToModelType: String,
                                   convertedModelPath: String
                                 ) extends Action {

  def this() = this(Forecast.MODEL_TYPE_MLEAP, null)

  val envIP: Map[String, String] = Map[String, String](
    CommonConstant.ENV_DEV -> "http://10.201.7.232:8063",
    CommonConstant.ENV_TEST -> "http://10.201.7.239:8063",
    CommonConstant.ENV_LIVE -> "http://sml.suishoudata.com"
  )

  override def act(data: DataFrame, validationParam: ValidationParam): Boolean = {
    // 检验运行环境，用于确定像那个 url 发送转换模型请求
    if (envIP.get(validationParam.env).isEmpty) {
      logError(s"检测运行环境失败，无法发送转换模型请求：未知环境：${validationParam.env}")
      return false
    }
    require(Forecast.SUPPORTED_MODEL_TYPES.contains(convertToModelType.toLowerCase), s"param convertToModelType only" +
      s" accepts [${Forecast.SUPPORTED_MODEL_TYPES.mkString(", ")}], but has $convertToModelType")

    logInfo(s"converting (${validationParam.sparkModels.mkString(", ")}) as $convertToModelType to $convertedModelPath")

    // 加上时间戳，防止报文件已存在错误
    val pathWithTime = ConvertModelTypeAction.modelPathTimeAppender(
      convertToModelType,
      Array(convertedModelPath),
      validationParam.flow_time)(0)
    var reqMap = Map[String, Any]()
    reqMap += ("sources" -> validationParam.sparkModels)
    reqMap += ("destination" -> pathWithTime)

    implicit val formats: DefaultFormats.type = DefaultFormats
    val body = write(reqMap)

    val path = convertToModelType.toLowerCase match {
      case Forecast.MODEL_TYPE_PMML =>
        "/web/ui/models/convert/spark-to-pmml"
      case Forecast.MODEL_TYPE_MLEAP =>
        "/web/ui/models/convert/spark-to-mleap"
    }
    val url = s"${envIP(validationParam.env)}$path"
    logInfo(s"posting to $url with body: $body")

    val response = HttpUtil.postWithEncryption(path, url, body)
    //val resMsg = read[Map[String, Any]](response)
    //val success = response.getStatusCode == 200 && resMsg("code") == 0
    val success = response.contains("200")

    if (success) {
      logInfo(s"转换 Spark 模型 [${validationParam.sparkModels.mkString(", ")}] 成 $convertToModelType 成功")
    } else {
      logError(s"转换 $convertToModelType 模型失败")
    }
    success
  }
}


object ConvertModelTypeAction {
  // 自动给填上的模型路径加上时间戳后缀
  def modelPathTimeAppender(modelType: String, originalPaths: Array[String], time: String): Array[String] = {
    modelType.toLowerCase match {
      case Forecast.MODEL_TYPE_SPARK =>
        originalPaths.map(_ + "_" + time)
      case Forecast.MODEL_TYPE_PMML | Forecast.MODEL_TYPE_MLEAP =>
        // pmml 和 mleap 模型都是单个带格式后缀的文件（.pmml .xml .zip）
        originalPaths.map(path => {
          val pathSeg = path.split("\\.")
          s"${pathSeg.dropRight(1).mkString(".")}_$time.${pathSeg.last}"
        })
    }
  }
}
